package com.diao.flink.streamapi.transform

import com.diao.flink.streamapi.source.SensorReading
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.scala._
/**
 * @author: Chenzhidiao
 * @date: 2020/3/27 14:09
 * @description:Union算子的应用，连接两条类型相同的流
 * @version: 1.0
 */
object Union {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    val stream1: DataStream[SensorReading] = env.readTextFile("E:\\Workspace\\flink\\src\\main\\resources\\unionFile1").map(data => {
      //将每条数据按逗号切分，然后封装成SensorReading对象
      val dataArray = data.split(",")
      SensorReading(dataArray(0).trim, dataArray(1).trim.toLong, dataArray(2).trim.toDouble)
    })

    val stream2: DataStream[SensorReading] = env.readTextFile("E:\\Workspace\\flink\\src\\main\\resources\\unionFile2").map(data => {
      //将每条数据按逗号切分，然后封装成SensorReading对象
      val dataArray = data.split(",")
      SensorReading(dataArray(0).trim, dataArray(1).trim.toLong, dataArray(2).trim.toDouble)
    })
    val unionStream: DataStream[SensorReading] = stream1.union(stream2)
    unionStream.print()

    env.execute()
  }
}
